package com.deman.framework.rpc.netty;

import com.deman.framework.rpc.config.RpcSystemConfig;
import com.deman.framework.rpc.config.ServiceNode;
import com.deman.framework.rpc.config.ServicesNodeManger;
import com.deman.framework.rpc.core.MessageCallBack;
import com.deman.framework.rpc.core.MessageCallBackHolder;
import com.deman.framework.rpc.model.MessageRequest;
import com.deman.framework.rpc.netty.handler.ClientInitChannel;
import com.deman.framework.rpc.properties.RpcSystemConfigProperties;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class ClientStarter implements InitializingBean {
    Logger logger= LoggerFactory.getLogger(ClientStarter.class);
    private EventLoopGroup workGroup = new NioEventLoopGroup();
    private static ClientStarter clientStarter = new ClientStarter();
    private Map <String, Boolean> mapRunning = new HashMap <>();
    private Map <String, Channel> channelMap = new ConcurrentHashMap <>();
    private Map <String, AtomicInteger> mapCount = new ConcurrentHashMap <>();
    @Autowired
    private RpcSystemConfigProperties rpcSystemConfigProperties;

    public static ClientStarter getInstance() {
        return clientStarter;
    }
    public void start(String serviceName, String host, int port) {
        this.connect(serviceName, host, port);
    }


    public Map <String, Boolean> getClientsStatus() {
        return mapRunning;
    }

    private void connect(String serviceName, String host, int port) {
        //2 辅助类(注意Client 和 Server 不一样)
        Bootstrap bootstrap = new Bootstrap();
        try {

            bootstrap.group(workGroup)
                    .channel(NioSocketChannel.class)
                    //表示缓存区动态调配（自适应）
                    .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                    //缓存区 池化操作
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .handler(new ClientInitChannel());
            //绑定端口，同步等等请求连接
            ChannelFuture channelFuture = bootstrap.connect(host, port);
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        future.channel().pipeline().get(ClientHandler.class);
                        System.err.println(" client started");
                        mapRunning.put(serviceName, true);
                    } else {
                        Thread.sleep(3000);
                        connect(serviceName, host, port);
                    }
                }
            });
            channelMap.put(serviceName, channelFuture.sync().channel());
        } catch (Exception e) {
            if(mapCount.get(serviceName)==null){
                mapCount.put(serviceName,new AtomicInteger(1));
            }else{
                mapCount.get(serviceName).addAndGet(1);
            }
            logger.error("{}第{}次链接失败[{}]",serviceName,mapCount.get(serviceName).get(),e.getMessage());

        }
    }


    public Object request(String serviceName, MessageRequest messageRequest) {
        messageRequest.setMessageId(UUID.randomUUID().toString());
        MessageCallBack messageCallBack = MessageCallBackHolder.put(messageRequest);
        channelMap.get(serviceName).writeAndFlush(messageRequest).addListener(new ChannelFutureListener(){
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if(future.isSuccess()){
                }
            }
        });
        Object obj = messageCallBack.start();
        return obj;
    }

    public static void main(String[] args) {

        ClientStarter client = new ClientStarter();
        client.start("test", "127.0.0.1", 765);

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Map <String, ServiceNode> map = ServicesNodeManger.build().getNodeMap();
        map.values().forEach(node->{
            if(StringUtils.isBlank(node.getIp())){
                node.setIp(rpcSystemConfigProperties.getServiceAddress());
            }
            if(node.getPort()==0){
                node.setPort(RpcSystemConfig.SERVICE_PORT);
            }
           this.start(node.getName(),node.getIp(),node.getPort());
        });
    }
}
